package com.ilikesoup.instant.executor;

import java.util.Collection;
import java.util.Collections;
import java.util.Set;
import java.util.concurrent.*;

/**
 * 该线程池实现支持，对于包含相同互斥键任务的有序执行。
 *
 * 注意：
 * 1 限于线程安全的实现方式，该线程池只能保证单个JVM的携带互斥键任务的有序执行，不能保证分布式系统的执行顺序。
 * 2 如果存在多个线程并行提交任务，且它们提交任务的互斥键可能存在冲突，那么要谨慎地选择等待队列的实现类，最好选用支持公平锁的实现。
 *
 * 基于以上两点，该类被设计用于单机串行模式的xxl-job使用。
 *
 * @author soup
 */
@SuppressWarnings("rawtypes")
public class KeyPickThreadPoolExecutor extends AbstractKeyPickThreadPoolExecutor {

    private final Set<Object> runningKeySet = Collections.newSetFromMap(new ConcurrentHashMap<>());

    @Override
    protected void removeKeys(Collection<?> keys) {
        runningKeySet.removeAll(keys);
    }

    /**
     * 只有单个线程控制 runningKeySet 的新增，即使其他线程在判断contains之后 addAll之前 完成了移除操作也没有关系。
     * @param keys
     * @return
     */
    @Override
    protected boolean addKeysIfNotExists(Collection<?> keys) {
        if(keys.stream().anyMatch(runningKeySet::contains)) {
            return false;
        }
        runningKeySet.addAll(keys);
        return true;
    }

    @Override
    public void clearKeys() {
        runningKeySet.clear();
    }

    private KeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    private KeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
    }

    private KeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
    }

    private KeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

    private KeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
                                      RejectedExecutionHandler handler, String scannerThreadName, long scannerThreadAwakeTimeout) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler, scannerThreadName, scannerThreadAwakeTimeout);
    }

    private KeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
                                      RejectedExecutionHandler handler, BlockingQueue<KeyPickRunnableFuture> waitingQueue, String scannerThreadName, long scannerThreadAwakeTimeout) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler, waitingQueue, scannerThreadName, scannerThreadAwakeTimeout);
    }

    private KeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, BlockingQueue<Runnable> workQueue, BlockingQueue<KeyPickRunnableFuture> waitingQueue, String scannerThreadName) {
        super(corePoolSize, maximumPoolSize, workQueue, waitingQueue, scannerThreadName);
    }

    public static KeyPickThreadPoolExecutor create(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        KeyPickThreadPoolExecutor executor = new KeyPickThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        executor.start();
        return executor;
    }

    public static KeyPickThreadPoolExecutor create(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        KeyPickThreadPoolExecutor executor = new KeyPickThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
        executor.start();
        return executor;
    }

    public static KeyPickThreadPoolExecutor create(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        KeyPickThreadPoolExecutor executor = new KeyPickThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        executor.start();
        return executor;
    }

    public static KeyPickThreadPoolExecutor create(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        KeyPickThreadPoolExecutor executor = new KeyPickThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        executor.start();
        return executor;
    }

    public static KeyPickThreadPoolExecutor create(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
                                                   RejectedExecutionHandler handler, String scannerThreadName, long scannerThreadAwakeTimeout) {
        KeyPickThreadPoolExecutor executor = new KeyPickThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler, scannerThreadName, scannerThreadAwakeTimeout);
        executor.start();
        return executor;
    }

    public static KeyPickThreadPoolExecutor create(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
                                                   RejectedExecutionHandler handler, BlockingQueue<KeyPickRunnableFuture> waitingQueue, String scannerThreadName, long scannerThreadAwakeTimeout) {
        KeyPickThreadPoolExecutor executor = new KeyPickThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler, waitingQueue, scannerThreadName, scannerThreadAwakeTimeout);
        executor.start();
        return executor;
    }

    public static KeyPickThreadPoolExecutor create(int corePoolSize, int maximumPoolSize, BlockingQueue<Runnable> workQueue, BlockingQueue<KeyPickRunnableFuture> waitingQueue, String scannerThreadName) {
        KeyPickThreadPoolExecutor executor = new KeyPickThreadPoolExecutor(corePoolSize, maximumPoolSize, workQueue, waitingQueue, scannerThreadName);
        executor.start();
        return executor;
    }

}
